package tasktracker.wordcount.managed;

import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import tasktracker.abstraction.TaskInfo;
import tasktracker.abstraction.TaskResultInfo;
import tasktracker.abstraction.TaskStatus;
import tasktracker.abstraction.extensions.ThreadedTaskTracker;

public class WordCountTaskTracker extends ThreadedTaskTracker {

	@Override
	protected void doThreadedTask(TaskInfo task) throws Exception {
		File inputBase = new File(task.getInputPath()).getAbsoluteFile();
		if (!inputBase.exists()) {
			reportStatus(task.getGuid(), TaskStatus.ERROR, new TaskResultInfo("No input files found."));
			return;
		}

		List<File> pendingFiles = new ArrayList<File>();
		for (File file : inputBase.listFiles()) {
			if (file.getName().toLowerCase().endsWith(".txt"))
				pendingFiles.add(file);
		}

		if (pendingFiles.isEmpty()) {
			reportStatus(task.getGuid(), TaskStatus.ERROR, new TaskResultInfo("No txt files found."));
			return;
		}

		File outputBase = new File(task.getOutputPath()).getAbsoluteFile();
		outputBase.mkdirs();

		Map<String, CountObject> map;
		List<Map.Entry<String, CountObject>> list;
		for (File file : pendingFiles) {
			logger.info("Task[" + task.getGuid() + "] Processing " + file.getName());
			map = process(file);
			list = sortResult(map);
			writeResultToFile(list, new File(outputBase, file.getName()));
			logger.info("Task[" + task.getGuid() + "] Fnished processing " + file.getName());
		}

		reportStatus(task.getGuid(), TaskStatus.COMPLETED, null);
	}

	private Map<String, CountObject> process(File file) {
		Map<String, CountObject> map = new HashMap<String, CountObject>();
		try (InputStream fileInputStream = new FileInputStream(file)) {
			try (InputStreamReader reader = new InputStreamReader(fileInputStream, "UTF-8")) {
				BufferedReader bufferedReader = new BufferedReader(reader);
				String lineTxt = null;
				while ((lineTxt = bufferedReader.readLine()) != null) {
					for (String part : lineTxt.split(" ")) {
						if ("".equals(part))
							continue;
						CountObject countObject = map.get(part);
						if (countObject == null) {
							countObject = new CountObject();
							map.put(part, countObject);
						}
						countObject.incerment();
					}
				}
			}
		} catch (IOException e) {
			logger.error("write to file failed.", e);
		}
		return map;
	}

	private List<Map.Entry<String, CountObject>> sortResult(Map<String, CountObject> result) {
		List<Map.Entry<String, CountObject>> entryList = new ArrayList<Map.Entry<String, CountObject>>(
				result.entrySet());
		Collections.sort(entryList, new Comparator<Map.Entry<String, CountObject>>() {
			@Override
			public int compare(Map.Entry<String, CountObject> o1, Map.Entry<String, CountObject> o2) {
				return (int) (o2.getValue().getCount() - o1.getValue().getCount());
			}
		});
		return entryList;
	}

	private void writeResultToFile(List<Map.Entry<String, CountObject>> result, File file) {
		try (OutputStream outputStream = new FileOutputStream(file, false)) {
			try (BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(outputStream)) {
				try (OutputStreamWriter writer = new OutputStreamWriter(bufferedOutputStream, "UTF-8")) {
					for (Map.Entry<String, CountObject> ele : result) {
						writer.write(ele.getKey() + "\t" + ele.getValue().getCount());
						writer.write('\n');
					}
				}
			}
		} catch (IOException e) {
			logger.error("write to file failed.", e);
		}
	}

}
